package net.java.nioserver.aio;

import net.java.nioserver.BasicService;
import net.java.nioserver.BasicServiceBuilder;
import net.java.nioserver.operations.IoEventListener;
import net.java.nioserver.utils.ChannelUtil;
import net.jcip.annotations.ThreadSafe;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.ServerSocket;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author Leonid Shlyapnikov
 */
@ThreadSafe
public class AioService extends BasicService {

    private static final Logger log = LoggerFactory.getLogger(AioService.class);

    protected final BasicAioOpAccept requiredOpAccept = new BasicAioOpAccept();

    // run-time fields

    private final ConcurrentLinkedQueue<SelectorChangeRequest> changeRequests = new ConcurrentLinkedQueue<SelectorChangeRequest>();

    private Selector selector;

    private volatile boolean isRunning = false;

    private AioService(Builder builder) {
        super(builder);
    }

    public boolean isRuning() {
        return isRunning;
    }

    public void start() throws IOException {
        init();
        runMainLoop();
    }

    public synchronized void stop(long millisToWait) {
        if (!isRunning) {
            log.info("The service is already shutdown " + this);
            return;
        }

        log.info("Shutting down " + this);

        isRunning = false;

        if (null != executor) {
//            executor.shutdownNow();
            executor.shutdown();
            try {
                executor.awaitTermination(millisToWait, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                log.warn("Interrupted while waiting for executor shutdown", e);
                Thread.currentThread().interrupt();
            }
        }

        // close all channels, cancel all keys
        for (SelectionKey key : selector.keys()) {
            ChannelUtil.closeChannelCancelKey(key);
        }

//        selector.wakeup();

        // close the selector
        try {
            selector.close();
        } catch (IOException e) {
            log.warn("Cannot close selector: " + selector, e);
        } finally {
            selector = null;
        }

        changeRequests.clear();
    }

    private void init() throws IOException {
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);
        ServerSocket ss = ssc.socket();
        ss.bind(address);

        selector = Selector.open();
        ssc.register(selector, SelectionKey.OP_ACCEPT);
    }

    private void runMainLoop() {
        isRunning = true;

        log.info("Started " + this);

        while (isRunning) {
            try {
                processChangeRequests();
                runOnePollTurn();
            } catch (Exception e) {
                log.error("Cannot finish poll turn", e);
            }
        }

        log.info("Stopping " + this + "...");
    }

    private void processChangeRequests() {
        for (SelectorChangeRequest changeRequest : changeRequests) {
            if (changeRequest.type() == SelectorChangeRequest.Type.CHANGEOPS) {
                SocketChannel socketChannel = changeRequest.socketChannel();
                if (!socketChannel.isOpen()) {
                    continue;
                }
                SelectionKey key = socketChannel.keyFor(this.selector);
                key.interestOps(changeRequest.ops());
            }
        }
    }

    void addChangeRequest(SelectorChangeRequest changeRequest) {
        changeRequests.add(changeRequest);
        selector.wakeup();
    }

    private void runOnePollTurn() throws IOException {
        int num = selector.select();

        if (0 == num || !isRunning) {
            return; // if it was woke up by selector.wakeup(), the number of selected keys can be zero, see addChangeRequest
        }

        // get all the keys/poll all events that need to be handled
        final Set<SelectionKey> selectedKeys = selector.selectedKeys();

        // process every event in the loop
        for (Iterator<SelectionKey> it = selectedKeys.iterator(); it.hasNext();) {
            if (!isRunning) {
                return;
            }

            final SelectionKey key = it.next();
            it.remove();

            if (!key.isValid()) {
                ChannelUtil.closeChannelCancelKey(key);
                continue;
            }

            if (key.isReadable()) {
                this.read(key);
            } else if (key.isWritable()) {
                this.write(key);
            } else if (key.isAcceptable()) {
                this.accept(key);
            }
        }
    }

    private void read(SelectionKey key) {
        SocketChannel sc = (SocketChannel) key.channel();
        if (!sc.isOpen()) {
            if (log.isDebugEnabled()) {
                log.debug("Channel closed: " + sc + ", skiping the event");
            }
            return;
        }
        dispatchEvent(opRead, sc, key);
    }

    private void write(SelectionKey key) {
        SocketChannel sc = (SocketChannel) key.channel();
        if (!sc.isOpen()) {
            if (log.isDebugEnabled()) {
                log.debug("Channel closed: " + sc + ", skiping the event");
            }
            return;
        }
        dispatchEvent(opRead, sc, key);
        //addChangeRequest(new SelectorChangeRequest(sc, SelectorChangeRequest.Type.CHANGEOPS, SelectionKey.OP_READ));
        key.interestOps(SelectionKey.OP_READ);
    }

    private void accept(SelectionKey key) throws IOException {
        // this should be done in the main thread, does not work from the thread pool
        ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
        SocketChannel sc = requiredOpAccept.handle(ssc, selector);

        if (null != opAccept) {
            // connection accepted, if we have a special listener, run it
            dispatchEvent(opAccept, sc, key);
        }
    }

    private void dispatchEvent(IoEventListener listener, SocketChannel sc, SelectionKey key) {
        if (!sc.isOpen()) {
            ChannelUtil.closeChannelCancelKey(key);
        }

        AioEventListenerTask command = new AioEventListenerTask(listener, sc, key);

        if (null != executor) {
            executor.execute(command); // use thread pool to execute a task asynchronously
        } else {
            command.run(); // run the task in the current thread synchronously
        }
    }

    public String toString() {
        return this.getClass().getSimpleName() + "{" +
                "address=" + address +
                ", executor=" + executor +
                ", opAccept=" + opAccept +
                ", opRead=" + opRead +
                ", isRunning=" + isRunning +
                '}';
    }

    public static class Builder extends BasicServiceBuilder {
        public BasicService build() {
            validate();
            return new AioService(this);
        }
    }
}

